package com.atguigu.windows;

import com.atguigu.bean.WaterSensor;
import com.atguigu.datasource.RandomWatersensor;
import com.atguigu.utils.ITtoListUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;
import java.util.List;

public class Flink08_WaterMark {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(1);

        env.addSource(new RandomWatersensor())
                //加水印
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        //指定水印最大容忍的延迟时间（无序状态）
                        .<WaterSensor> forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        //指定那个字段作为水印比较时间戳
                        .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
                )
                //进行keyBy操作
                .keyBy(new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor value) throws Exception {
                        return value.getId();
                    }
                })
                //进行窗口处理
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //对数据进行处理
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context context,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                        String start = new Date(context.window().getStart()).toLocaleString();
                        String end = new Date(context.window().getEnd()).toLocaleString();
                        System.out.println("窗口起始："+start+",窗口结束："+end);
                        //输出元素
                        List<WaterSensor> list = ITtoListUtil.toList(elements);

                        //元素原样输出
                        out.collect(list.toString());
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
